"""
    helper.py
    ~~~~~~~~
    助手函数集

    :author: Fufu, 2019/9/9
    :update: Fufu, 2020/9/7 get_date() 增加支持时间戳(秒/毫秒)
    :update: Fufu, 2020/12/16 get_iso_date()
"""
import asyncio
import calendar
import hashlib
import json
import math
import os
import re
import socket
import sys
import time
import uuid
from datetime import datetime, timedelta, date
from decimal import Decimal
from functools import wraps, lru_cache
from inspect import iscoroutinefunction
from itertools import chain
from threading import Thread
from typing import Any, Optional, Union, List
import requests
from loguru import logger
from requests.adapters import HTTPAdapter
from cryptography.fernet import Fernet
from urllib.parse import quote, unquote
from Crypto.Cipher import AES
from Crypto.Util import Counter
import base64


def try_logger(
        depth: int = 1,
        *,
        as_logger: bool = True,
        log_level: str = 'info',
        log_tag: str = '',
        retries: int = 0,
        backoff_sec: int = 1,
        throw: bool = False,
):
    """
    带 Logger 的 try

    e.g.::

        @try_logger()
        def foo():
            1/0

        @try_logger()
        async def foo():
            1/0

    :param depth: int, 日志记录调用器深度
    :param as_logger: bool, 是否记录日志
    :param log_level: str, 日志级别, 默认 'info'
    :param log_tag: str, 日志标识文本
    :param retries: 重试次数, 0 则只执行一次
    :param backoff_sec: 重试时等待的基础秒数
    :param throw: bool, 是否抛出异常
    :return:
    """
    if retries <= 0:
        retries = 0

    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            log = logger.opt(depth=depth)
            log = getattr(log, log_level, log.info)
            msg = '{} - {}'.format(log_tag, fn.__name__) if log_tag else fn.__name__
            for i in range(1 if retries <= 0 else retries + 1):
                try:
                    if as_logger:
                        log('try {} start'.format(msg))
                        start = time.perf_counter()
                        res = fn(*args, **kwargs)
                        cost = time.perf_counter() - start
                        log('try {} end, cost: {:.6f}s'.format(msg, cost))
                    else:
                        res = fn(*args, **kwargs)
                    return res
                except Exception as e:
                    as_logger and log.opt(exception=True).error('try {} error: {}'.format(msg, e))
                    if throw:
                        raise e
                    i < retries and time.sleep(backoff_sec + i)

        @wraps(fn)
        async def async_wrapper(*args, **kwargs):
            log = logger.opt(depth=depth)
            log = getattr(log, log_level, log.info)
            msg = '{} - {}'.format(log_tag, fn.__name__) if log_tag else fn.__name__
            for i in range(1 if retries <= 0 else retries + 1):
                try:
                    if as_logger:
                        log('try {} start'.format(msg))
                        start = time.perf_counter()
                        res = await fn(*args, **kwargs)
                        cost = time.perf_counter() - start
                        log('try {} end, cost: {:.6f}s'.format(msg, cost))
                    else:
                        res = await fn(*args, **kwargs)
                    return res
                except Exception as e:
                    as_logger and log.opt(exception=True).error('try {} error: {}'.format(msg, e))
                    if throw:
                        raise e
                    i < retries and time.sleep(backoff_sec + i)

        return async_wrapper if iscoroutinefunction(fn) else wrapper

    return decorator


def run_perf(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        res = fn(*args, **kwargs)
        end = time.perf_counter()
        print('\n~~~~~~\n{}.{} : {}\n~~~~~~\n'.format(fn.__module__, fn.__name__, end - start))
        return res

    return wrapper


def debug(*data, end='------\n', die=False):
    """输出调试信息"""
    from pprint import pprint
    for x in data:
        pprint(x)
    print('', end=end)
    die and sys.exit(1)


def get_plain_text(s):
    """
    获取纯文本内容
    清除 html 标签, 空白字符替换为一个空格, 去除首尾空白

    :param s: str
    :return: str
    """
    return re.sub(r'\s+', ' ', re.sub(r'<[^>]+>', '', s)).strip()


def get_uniq_list(data):
    """
    列表去重并保持顺序(数据量大时效率不高)

    :param data: list
    :return: list
    """
    ret = list(set(data))
    ret.sort(key=data.index)
    return ret


def list2dict(key, *value):
    """
    列表转换列表.字典(用于数据记录转字典列表)

    e.g.::

        list2dict(('a', 'b'), [(1, 2), (3, 4)], [('x', 22)])

    :param key: tuple, list, 元组或列表, 与每行数据元素个数相同
    :param value: list.tuple, 一组元组数据
    :return: list.dict
    """
    try:
        return [dict(zip(key, x)) for x in chain(*value)]
    except Exception:
        return []


def rows2kv(rows, key_name: str, value_name: str = '') -> dict:
    """
    数据库结果集转键值对

    e.g.::

        # {23646: 'PUBG', ...}
        rows2kv(games_rows, 'game_id', 'game_name')

    :param rows: 数据集
    :param key_name: str, 用于字典键名的字段名
    :param value_name: str, 为空时与 key_name 相同
    :return:
    """
    if not value_name:
        value_name = key_name
    try:
        return {x[key_name]: x[value_name] for x in rows}
    except Exception:
        return {}


def data2list(data, sep=','):
    """
    字符串转列表

    :param data: 非字符串和列表外, 其他类型返回空列表
    :param sep: 字符串分隔符
    :return: list
    """
    if isinstance(data, list):
        return data
    elif data and isinstance(data, str):
        return data.split(sep)
    else:
        return []


def dict2obj(dic):
    """
    字典转对象
    :param dic:
    :return: obj
    """

    obj = type('new', (object,), dic)
    seqs = tuple, list, set, frozenset
    for key, value in dic.items():
        if isinstance(value, dict):
            setattr(obj, key, dict2obj(value))
        elif isinstance(value, seqs):
            setattr(obj, key, type(value)(dict2obj(x) if isinstance(x, dict) else x for x in value))
        else:
            setattr(obj, key, value)
    return obj


def get_round(s=None, default=None, precision=2, sep=None):
    """
    检查是否为浮点数, 转换(四舍六入五成双)并返回 float 或 列表

    e.g.::

        # None
        get_round('1a')

        # 0
        get_round('1a', 0)

        # 123.0
        get_round(' 123   ')

        # [123.0, 456.34, 7.0]
        get_round('123, 456.336,, 7,, ,', sep=',')

    :param s: str, 整数值或字符串
    :param default: 转换整数失败时的默认值(列表转换时无效)
    :param precision: 小数位置, 默认 2 位
    :param sep: str, 指定分隔符时返回 list, e.g. [1, 7]
    :return: int / list / default
    """
    if isinstance(s, (float, int, Decimal)):
        return round(float(s), precision)
    elif isinstance(s, str):
        s = s.strip()
        try:
            if sep:
                ret = [round(float(x), precision) for x in s.split(sep) if x.strip() != '']
                ret = get_uniq_list(ret)
            else:
                ret = round(float(s), precision)
        except ValueError:
            ret = [] if sep else default
    else:
        ret = [] if sep else default

    return ret


def get_int(s=None, default=None, sep=None):
    """
    检查是否为整数, 转换并返回 int 或 列表

    e.g.::

        # None
        get_int('1a')

        # 0
        get_int('1a', 0)

        # 123
        get_int(' 123   ')

        # [123, 456, 7]
        get_int('123, 456,, 7,, ,', sep=',')

    :param s: str, 整数值或字符串
    :param default: 转换整数失败时的默认值(列表转换时无效)
    :param sep: str, 指定分隔符时返回 list, e.g. [1, 7]
    :return: int / list / default
    """
    if isinstance(s, bytes):
        try:
            s = s.decode('utf-8')
        except Exception:
            return [] if sep else default

    if isinstance(s, int):
        return int(s)
    elif isinstance(s, (float, Decimal)):
        return int(float(s))
    elif isinstance(s, str):
        s = s.strip()
        try:
            if sep:
                ret = [int(x) for x in s.split(sep) if x.strip() != '']
                ret = get_uniq_list(ret)
            else:
                ret = int(s)
        except ValueError:
            ret = [] if sep else default
    else:
        ret = [] if sep else default

    return ret


def get_domain(domain=None):
    """检查是否为域名, 清除域名前后空白后返回"""
    domain = domain.strip()
    patt = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
    return domain if re.match(patt, domain) else None


def get_date(any_dt=None, in_fmt='%Y-%m-%d', out_fmt='', default=True,
             add_days=0, add_hours=0, add_minutes=0, add_seconds=0):
    """
    检查日期是否正确并返回日期

    e.g.::

        get_date('2018-10-10')
        get_date('2018-10-10 12:00:00', '%Y-%m-%d %H:%M:%S')
        get_date('1599444062')
        get_date(out_fmt='timestamp000')
        get_date(out_fmt='date')
        get_date(out_fmt='datetime')
        get_date(out_fmt='begin_of_minute')
        get_date(date(year=2020, month=1, day=1))
        get_date(date(year=2020, month=1, day=1), out_fmt='datetime')

        # '2021-11-05T14:00:00+08:00'
        get_date('2021-11-05T14:00:00.283+08:00', in_fmt='iso', out_fmt='iso')
        get_date('2021-11-05T14:00:00.000Z', in_fmt='iso', out_fmt='iso')
        get_date('2021-11-05T14:15:16.999Z', in_fmt='iso', out_fmt='begin_of_minute')

        # '2017-12-10 12:04:00'
        get_date('2017-12-10T04:04Z', in_fmt='iso.aly', out_fmt='%Y-%m-%d %H:%M:%S')


    :param any_dt: mixed, 输入的日期, 空/日期字符串/日期对象/时间戳
    :param in_fmt: str, 源日期格式
    :param out_fmt: str, 返回的日期格式, 默认返回日期对象, 特殊: timestamp 时返回时间戳, timestamp000 毫秒时间戳
    :param default: bool, True 源日期格式不正确时返回今天
    :param add_days: int, 正负数, 与输入日期相差的天数
    :param add_hours: int, 正负数, 与输入日期相差的小时数
    :param add_minutes: int, 正负数, 与输入日期相差的分钟数
    :param add_seconds: int, 正负数, 与输入日期相差的秒数
    :return: datetime|None|str
    """
    dt = None
    if isinstance(any_dt, datetime):
        dt = any_dt
    elif isinstance(any_dt, date):
        dt = datetime(year=any_dt.year, month=any_dt.month, day=any_dt.day)
    else:
        try:
            aly_hours = 0
            in_fmt = str(in_fmt)
            if in_fmt.startswith('iso'):
                if 'aly' in in_fmt:
                    # 阿里云接口返回的时间: '2017-12-10T04:04Z', 转换为北京时间
                    in_fmt = '%Y-%m-%dT%H:%M'
                    aly_hours = 8
                else:
                    in_fmt = '%Y-%m-%dT%H:%M:%S'
                any_dt = str(any_dt).strip()
                if '.' in any_dt:
                    in_fmt += '.%f'
                if 'Z' in any_dt:
                    in_fmt += 'Z'
                elif '+' in any_dt:
                    tmp = any_dt.split('+')
                    if len(tmp) < 2:
                        any_dt = tmp[0]
                    else:
                        any_dt = tmp[0] + '+' + tmp[1].replace(':', '')
                        in_fmt += '%z'
            # 注: python3 会丢弃时区信息
            dt = datetime.strptime(any_dt, in_fmt) + timedelta(hours=aly_hours)
        except Exception:
            timestamp = get_int(any_dt)
            if timestamp:
                try:
                    dt = datetime.fromtimestamp(timestamp)
                except Exception:
                    pass

    if not dt:
        if default:
            dt = datetime.now()
        else:
            return None

    if add_days and isinstance(add_days, int):
        dt = dt + timedelta(days=add_days)

    if add_hours and isinstance(add_hours, int):
        dt = dt + timedelta(hours=add_hours)

    if add_minutes and isinstance(add_minutes, int):
        dt = dt + timedelta(minutes=add_minutes)

    if add_seconds and isinstance(add_seconds, int):
        dt = dt + timedelta(seconds=add_seconds)

    if not out_fmt:
        return dt
    if out_fmt == 'iso':
        return get_iso_date(dt)
    if out_fmt == 'timestamp000':
        return int(dt.timestamp() * 1000)
    if out_fmt == 'timestamp':
        return int(dt.timestamp())
    if out_fmt == 'datetime':
        if not getattr(dt, 'date', None):
            dt = datetime(dt.year, dt.month, dt.day)
        return dt
    if out_fmt == 'date':
        return dt.date() if getattr(dt, 'date', None) else dt
    if out_fmt == 'begin_of_minute':
        return datetime(dt.year, dt.month, dt.day, dt.hour, dt.minute)
    if out_fmt == 'begin_of_hour':
        return datetime(dt.year, dt.month, dt.day, dt.hour)
    if out_fmt == 'begin_of_day':
        return datetime(dt.year, dt.month, dt.day)
    if out_fmt == 'begin_of_month':
        return datetime(dt.year, dt.month, 1)
    if out_fmt == 'interval_5m':
        return dt.replace(minute=(dt.minute // 5) * 5, second=0, microsecond=0)

    return dt.strftime(out_fmt)


def get_iso_date(any_dt=None, in_fmt='%Y-%m-%d', zone='+08:00'):
    """
    世界时间格式, 默认当前时间
    注: 忽略输入的时区, 结果强制为指定的时区

    e.g.::

        get_iso_date()
        get_iso_date('2020-01-01')
        get_iso_date(date(year=2020, month=1, day=1))
        get_iso_date('2020-01-01 01:02:03', in_fmt='%Y-%m-%d %H:%M:%S')

        # '2021-11-05T14:00:00Z'
        get_iso_date('2021-11-05T14:00:00.000Z', in_fmt='iso', zone='Z')
        get_iso_date('2021-11-05T14:00:00.283+08:00', in_fmt='iso', zone='Z')

    :param any_dt: mixed, 输入的日期, 空/日期字符串/日期对象/时间戳
    :param in_fmt: str, 源日期格式
    :param zone:
    :return:
    """
    return get_date(any_dt, in_fmt=in_fmt, out_fmt='datetime').isoformat(timespec='seconds').split('+')[0] + zone


def get_date_range(sdate=None, edate=None, days=None, in_fmt='%Y-%m-%d', out_fmt='', with_edate=True):
    """
    获取日期范围

    e.g.::

        # [datetime.datetime(2020, 12, 14, 11, 52, 37, 540895)]
        list(get_date_range()

        # ['20201212', '20201213']
        list(get_date_range('2020-12-12', '2020-12-13', out_fmt='%Y%m%d'))

        # ['20201212', '20201213']
        list(get_date_range('2020-12-12', out_fmt='%Y%m%d', with_edate=False))

    :param sdate: 开始时间
    :param edate: 结束时间
    :param days: 时间间隔
    :param in_fmt: str, 源日期格式
    :param out_fmt: str, 返回的日期格式, 默认返回日期对象, 特殊: timestamp 时返回时间戳
    :param with_edate: bool, 是否包含结束日期
    :return: list
    """
    sdate = get_date(sdate, in_fmt=in_fmt)
    edate = get_date(edate, in_fmt=in_fmt)

    if not isinstance(days, timedelta):
        days = timedelta(days=1)

    for i in range((edate - sdate).days + (1 if with_edate else 0)):
        yield get_date(sdate + days * i, out_fmt=out_fmt)


def get_ymd(dt=None, in_fmt='%Y-%m-%d', out_fmt='%Y-%m-%d', default=True, add_days=0):
    """方便获取年月日格式的日期"""
    return get_date(dt, in_fmt=in_fmt, default=default, add_days=add_days, out_fmt=out_fmt)


def get_next_month_first(dt=None, in_fmt='%Y-%m-%d', out_fmt=''):
    """
    获取日期的下月第一天

    e.g.::

        # 2020-07-01
        get_next_month_first('2020-06-28', out_fmt='%Y-%m-%d')

    :param dt: mixed, 输入的日期, 空/日期字符串/日期对象
    :param in_fmt: str, 源日期格式
    :param out_fmt: str, 返回的日期格式, 默认返回日期对象
    :return: datetime|None|str
    """
    dt = get_date(dt, in_fmt=in_fmt)
    return get_date(datetime(dt.year, dt.month, 1) + timedelta(days=calendar.monthrange(dt.year, dt.month)[1]),
                    out_fmt=out_fmt)


def get_month_last(dt=None, in_fmt='%Y-%m-%d', out_fmt=''):
    """
    获取当月最后一天

    e.g.::

        # 200630 235959
        get_month_last('2020-06-12', out_fmt='%y%m%d %H%M%S')

    :param dt: mixed, 输入的日期, 空/日期字符串/日期对象
    :param in_fmt: str, 源日期格式
    :param out_fmt: str, 返回的日期格式, 默认返回日期对象
    :return: datetime|None|str
    """
    dt = get_next_month_first(dt, in_fmt=in_fmt) + timedelta(seconds=-1)
    return get_date(dt, out_fmt=out_fmt)


def get_last_month_last(dt=None, in_fmt='%Y-%m-%d', out_fmt=''):
    """
    获取上月最后一天

    e.g.::

        # 200531 235959
        get_last_month_last('2020-06-12', out_fmt='%y%m%d %H%M%S')

    :param dt: mixed, 输入的日期, 空/日期字符串/日期对象
    :param in_fmt: str, 源日期格式
    :param out_fmt: str, 返回的日期格式, 默认返回日期对象
    :return: datetime|None|str
    """
    dt = get_date(dt, in_fmt=in_fmt)
    return get_date(datetime(dt.year, dt.month, 1) + timedelta(seconds=-1), out_fmt=out_fmt)


def get_hash(data=None, hash_name='md5', salt=''):
    """
    获取数据的摘要信息

    :param data: str, list, dict...
    :param hash_name: str, e.g. 'md5', 'sha1', 'sha224', 'sha256'...
    :param salt: str
    :return:
    """
    try:
        m = getattr(hashlib, hash_name)(salt if isinstance(salt, bytes) else bytes(str(salt), 'utf-8'))
        m.update(data if isinstance(data, bytes) else bytes(str(data), 'utf-8'))
        return m.hexdigest()
    except Exception:
        return ''


def get_large_file(url, file=None, chunk_size=4096, retries=3, timeout=(30, 30), throw=False, **kwargs):
    """
    下载大文件(流), 默认重试 3 次
    保存到文件或返回 requests 对象

    :param url: str, 下载链接
    :param file: str, 结果保存文件
    :param chunk_size: int, 文件保存时的块大小
    :param retries: int, 重试次数
    :param timeout: int | tuple 连接超时和读取超时
    :param throw: bool, 是否抛出异常
    :param kwargs: dict, requests 自定义参数
    :return: bool | requests
    """
    s = requests.Session()
    s.mount('http://', HTTPAdapter(max_retries=retries))
    s.mount('https://', HTTPAdapter(max_retries=retries))

    try:
        r = s.get(url, stream=True, timeout=timeout, **kwargs)
        if file:
            try:
                with open(file, 'wb+') as f:
                    for chunk in r.iter_content(chunk_size):
                        f.write(chunk)
                return True
            except Exception as e:
                if throw:
                    raise e
                return False
        return r
    except requests.exceptions.RequestException as e:
        if throw:
            raise e
        return False


def req(
        method: str,
        url: str,
        *,
        retries: int = 3,
        retry_status: Optional[Union[list, tuple, set]] = None,
        backoff_sec: int = 1,
        timeout: Optional[Union[int, tuple]] = 10,
        as_json: bool = True,
        as_headers: bool = False,
        as_logger: bool = True,
        **kwargs,
) -> Union[tuple, dict, str]:
    """
    发起请求, 默认重试 3 次, 支持返回头信息
    注: 默认解析 JSON 结果

    e.g.::

        txt = req('GET', 'http://4.ipw.cn', as_json=False)
        js = req('GET', 'http://ipinfo.io')
        js = req('GET', 'https://ipinfo.io', as_headers=True)
        js = req('POST', 'http://192.168.1.100:555/code/504', as_headers=True, retries=2)

    :param method: 显式指定请求方式: GET POST PUT ...
    :param url: 请求地址
    :param retries: 重试次数
    :param retry_status: 需要重试的响应状态码
    :param backoff_sec: 重试时等待的基础秒数
    :param timeout: 请求超时, 默认 10 秒, 可用元组指定读写超时
    :param as_json: 默认解析 JSON 结果
    :param as_headers: 可选返回响应头信息
    :param kwargs: 其他 requests 可选参数, 如: json=dict_data
    :param as_logger: 默认打印日志
    :return:
    """
    if retries <= 0:
        retries = 0
        retry_status = []
    else:
        if not isinstance(retry_status, (list, tuple, set)):
            retry_status = [413, 429, 503, 504]
    with requests.Session() as s:
        for i in range(1 if retries <= 0 else retries + 1):
            try:
                r = s.request(method, url, timeout=timeout, **kwargs)
                as_logger and logger.opt(depth=1).debug(
                    f'{i}. [{r.status_code}:{r.elapsed.total_seconds():.03f}s] {method} {url}')
                if r.status_code in retry_status:
                    i < retries and time.sleep(backoff_sec + i)
                    continue
                if as_json:
                    return (r.json(), r.headers) if as_headers else r.json()
                else:
                    return (r.text, r.headers) if as_headers else r.text
            except Exception as e:
                as_logger and logger.opt(depth=1).error(f'{i}. {method} {url} {e}')
                i < retries and time.sleep(backoff_sec + i)
        if as_json:
            return ({}, {}) if as_headers else {}
        else:
            return ('', {}) if as_headers else ''


def async_task(fn):
    """
    装饰器: 异步执行任务

    e.g.::

        @async_task
        def _async_send_mail():
            pass

    :param fn:
    :return:
    """

    @wraps(fn)
    def wrapper(*args, **kwargs):
        t = Thread(target=fn, args=args, kwargs=kwargs)
        t.start()

    return wrapper


def get_json_loads(s, default=False, *, as_file=False):
    """屏蔽错误, 加载 JSON 字符串或文件"""
    if isinstance(s, dict):
        return s
    try:
        if as_file:
            with open(s, 'r') as f:
                return json.load(f)
        else:
            return json.loads(s)
    except Exception:
        return {} if default is False else default


def get_dict_value(di, key_path, default=None, *, as_true=False, fix_type=True):
    """
    获取字典键值
    源字典类型错误时, 返回默认值
    !!! 键路径为空时, 返回源字典
    获取的结果为假时, 可选返回默认值
    获取的结果类型自动转换为默认值相同类型, 转换失败返回默认值, 默认值为 None 时不转换

    e.g.::

        # {'a': {'b': 0}}
        get_dict_value({'a': {'b': 0}}, '', 1)

        # 1
        get_dict_value({'a': {'b': 0}}, 'a', 1)

        # {'b':0}
        get_dict_value({'a': {'b': 0}}, 'a', {})

        # 0
        get_dict_value({'a': {'b': 0}}, 'a|b', 0)

        # 1
        get_dict_value({'a': {'b': 0}}, 'a|b', 1, as_true=True)

        # 0
        get_dict_value({'a': {'b': 0}}, '  a  |||  b', 1)

        # 0
        get_dict_value({'a': {'b': 0}}, '  a  |||  b', 1, as_true=False)

        # 1
        get_dict_value({'a': {'b': 0}}, 'a|b|c', 1)

        # 1
        get_dict_value({'a':True}, 'a', 0)

        # True
        get_dict_value({'a':True}, 'a', False)

        # True
        get_dict_value({'a':True}, 'a', 0, fix_type=False)

        # 0
        get_dict_value({'a':False}, 'a', 0)

        # False
        get_dict_value({'a':False}, 'a',False)

        # False
        get_dict_value({'a':False}, 'a', 0, as_true=False, fix_type=False)

        # 0
        get_dict_value({'a':False}, 'b', 0)

        # 0
        get_dict_value(1, 'b', 0)

        # None
        get_dict_value({'a':False}, 'b')

    :param di: dict
    :param key_path: e.g. x|new_name
    :param default:
    :param as_true: bool, 字典值为假是返回默认值
    :param fix_type: 是否强制修正为 default 相同类型
    :return:
    """
    if not isinstance(di, dict):
        return default

    key_path = str(key_path).strip()
    if key_path == '':
        return di

    value = di
    for key in key_path.split('|'):
        key = key.strip()
        if key == '':
            continue
        try:
            value = value.get(key, None)
            if value is None:
                return default
        except Exception:
            return default

    if as_true and not value:
        return default

    if fix_type and default is not None:
        return get_same_type(default, value)

    return value


def get_same_type(src, dst):
    """将 dst 的类型转换为 src 相同的类型"""
    if src is None:
        return dst

    if isinstance(src, bool):
        return get_bool(dst)

    try:
        return type(src)(dst)
    except Exception:
        return src


def get_bool(v: Any) -> bool:
    """转换为布尔值"""
    return str(v) in ('1', 't', 'T', 'true', 'TRUE', 'True')


def get_split(v: Any, *, sep: str = ',', skip: List[str] = None) -> list:
    """分隔字符串, 清除空白"""
    skip = list(skip) if skip else []
    ret = []
    for d in str(v).split(sep):
        d = d.strip()
        d and d not in skip and ret.append(d)
    return ret


def is_json(_json):
    """判断是否为json"""
    try:
        json.loads(_json)
    except ValueError as e:
        return False
    return True


def query_to_sql(query):
    """
    打印原始Sql语句
    :param query:
    :return:
    """
    sql = str(query.compile(compile_kwargs={"literal_binds": True}))
    return sql


def create_directory_if_not_exists(path):
    """
    判断目录是否存在 不存在则创建
    """
    if not os.path.exists(path):
        os.makedirs(path)


def desensitize_name(name):
    # 将姓氏替换为"*"
    last_name = name[0]
    desensitized_name = last_name + "*" * (len(name) - 1)
    return desensitized_name


def desensitize_number(number):
    """
    数字中间4位脱敏
    """
    num_len = len(number)
    if num_len <= 4:
        return "****"

    # 计算脱敏位置
    start_index = (num_len - 4) // 2
    end_index = start_index + 4

    # 将中间四位替换为脱敏部分
    desensitized_number = number[:start_index] + "*" * 4 + number[end_index:]
    return desensitized_number


def float_round(num, digit=4):
    """
    进行四舍五入 保留4位小数
    :param num:
    :param digit:
    :return:
    """
    if isinstance(num, int) or (isinstance(num, str) and num.isdigit()):
        return num
    formatted_num = "{:.{}f}".format(float(num), digit)
    return formatted_num


def truncate(number, digit=4):
    """
    不进行四舍五入 保留4位小数
    :param number:
    :param digit:
    :return:
    """
    if isinstance(number, int) or (isinstance(number, str) and number.isdigit()) or number == int(number):
        return number
    factor = 10.0 ** digit
    truncated_number = math.trunc(float(number) * factor) / factor
    formatted_num = "{:.{}f}".format(float(truncated_number), digit)
    return formatted_num


def encrypt_data(data: str, key: str):
    # 创建加密器
    cipher = Fernet(key)
    # 加密数据
    encrypted_data = cipher.encrypt(data.encode('utf-8'))
    return encrypted_data.decode('utf-8')


def decrypt_data(data: str, key: str):
    # 创建加密器
    cipher = Fernet(key)
    # 加密数据
    decrypted_data = cipher.decrypt(data.encode('utf-8'))
    return decrypted_data.decode('utf-8')


def generate_uuid():
    # 生成一个UUID对象
    uuid_obj = uuid.uuid4()

    # 将UUID转换为32位的字符串并去除短横线
    uuid_str = uuid_obj.hex.replace('-', '')

    return uuid_str


async def retry_async_function(async_func, max_retries=5, delay=3, *args, **kwargs):
    attempts = 0
    while attempts < max_retries:
        try:
            return await async_func(*args, **kwargs)
        except Exception as e:
            attempts += 1
            if attempts == max_retries:
                raise e
            await asyncio.sleep(delay)


def get_enum_key_by_value(enum_class, value):
    for key, member in enum_class.__members__.items():
        if member.value == value:
            return key


def format_float(number):
    formatted = "{:.2f}".format(number)
    return formatted.rstrip('0').rstrip('.')


def truncate_string(string, limit):
    """
    截取字符串
    """
    if len(string) <= limit:
        return string
    else:
        return string[:limit - 3] + '...'


def aes_ctr_encrypt(key, iv, plaintext):
    key = key.encode('utf-8')
    iv = iv.encode('utf-8')
    plaintext = plaintext.encode('utf-8')

    # 创建 Counter 对象
    counter = Counter.new(AES.block_size * 8, initial_value=int.from_bytes(iv, byteorder='big'))

    # 创建 AES 加密对象
    cipher = AES.new(key, AES.MODE_CTR, counter=counter)

    # 加密明文
    ciphertext = cipher.encrypt(plaintext)

    # 返回密文
    return quote(base64.b64encode(ciphertext).decode('utf-8'))


def aes_ctr_decrypt(key, iv, ciphertext):
    # 从密文中提取 IV
    key = key.encode('utf-8')
    iv = iv.encode('utf-8')
    ciphertext = unquote(ciphertext)
    ciphertext = ciphertext.encode('utf-8')

    # 创建 Counter 对象
    counter = Counter.new(AES.block_size * 8, initial_value=int.from_bytes(iv, byteorder='big'))

    # 创建 AES 解密对象
    cipher = AES.new(key, AES.MODE_CTR, counter=counter)

    ciphertext = base64.b64decode(ciphertext)

    # 解密密文
    plaintext = cipher.decrypt(ciphertext)

    # 返回明文
    return plaintext.decode("utf-8")


def format_number(num, is_decimal=True):
    # 保留两位小数 并设置千分位
    if not is_decimal:
        return '{:,.0f}'.format(num)
    formatted_num = "{:,.2f}".format(num)
    if "." not in formatted_num:
        formatted_num += ".00"
    return formatted_num


def get_95th(datas: list) -> float:
    """返回传统 95 值(从高到低, 取前 5% 之后的那个值)"""
    if not datas or not isinstance(datas, list):
        return 0

    n = len(datas)
    if n == 1:
        return datas[0]

    sorted_datas = sorted(datas, reverse=True)
    return sorted_datas[int(0.05 * n):][0]


def get_e95th_daily(datas: list) -> float:
    """返回增强 95 日峰值(从高到低取第 5 个值, 小于 5 个值时取最大值)"""
    if not datas or not isinstance(datas, list):
        return 0

    if len(datas) < 5:
        return max(datas)

    sorted_datas = sorted(datas, reverse=True)
    return sorted_datas[4]


def get_e95th(datas: list) -> float:
    """返回增强 95 值(从高到低取前 5 个值的平均值, 小于 5 个直接取平均值)"""
    if not datas or not isinstance(datas, list):
        return 0

    n = len(datas)
    if n <= 5:
        return get_round(sum(datas) / n)

    sorted_datas = sorted(datas, reverse=True)
    e95 = get_round(sum(sorted_datas[:5]) / 5)
    return e95


def get_hostname():
    """获取主机标识"""
    try:
        name = socket.gethostname()
    except Exception:
        name = str(read_file('/etc/hostname')).strip()
    return name


def read_file(file_path, *, encoding='utf-8'):
    """读取文件内容"""
    try:
        with open(file_path, 'r', encoding=encoding) as f:
            content = f.read()
        return content
    except Exception as e:
        logger.opt(depth=1).error(f"failed to read file: {e}")
        return None


@lru_cache()
def get_ip_location(oper_ip: str):
    """
    查询ip归属区域

    :param oper_ip: 需要查询的ip
    :return: ip归属区域
    """
    oper_location = '内网IP'
    try:
        if oper_ip != '127.0.0.1' and oper_ip != 'localhost':
            oper_location = '未知'
            ip_result = requests.get(f'https://qifu-api.baidubce.com/ip/geo/v1/district?ip={oper_ip}')
            if ip_result.status_code == 200:
                prov = ip_result.json().get('data').get('prov')
                city = ip_result.json().get('data').get('city')
                if prov or city:
                    oper_location = f'{prov}-{city}'
    except Exception as e:
        oper_location = '未知'
        print(e)
    return oper_location